CodSpeed Performance Report

Merging #3545 will degrade performances by 13.84%

_{Comparing colin/incremental-metadata-retrieval (f15c7e5) with main (6157ad8)}

Summary

❌ 1 regressions
✅ 26 untouched benchmarks

⚠️ Please fix the performance issues or acknowledge them on CodSpeed.

Benchmarks breakdown

Codecov Report

Attention: Patch coverage is 96.42857% with 1 line in your changes missing coverage. Please review.

Project coverage is 77.17%. Comparing base (c932ec9) to head (f15c7e5).
Report is 23 commits behind head on main.

@@            Coverage Diff             @@
##             main    #3545      +/-   ##
==========================================
- Coverage   78.06%   77.17%   -0.90%     
==========================================
  Files         728      728              
  Lines       89967    91917    +1950     
==========================================
+ Hits        70236    70936     +700     
- Misses      19731    20981    +1250     

Interesting. One suggestion I have is to maybe see if we can use ray.wait([...metadatas], fetch_local=True) (see: https://docs.ray.io/en/latest/ray-core/api/doc/ray.wait.html)

I wonder if this might simplify the logic here, and it might also fix some interesting OOM issues that I was observing wrt the workers that are holding the metadata objectrefs dying due to OOM.

Could we try that and see if it gives us the same effect? We have to figure out where is appropriate to call this on the metadata objectrefs though. I would look into _await_tasks on our RayRunner.

Another idea... Looks like these metadatas are being retrieved using a Ray remote function:

@ray.remote
def get_metas(*partitions: MicroPartition) -> list[PartitionMetadata]:
    return [PartitionMetadata.from_table(partition) for partition in partitions]

My guess is that if the MicroPartition lives in the object store, we're having to page this data back in just to grab metadata. This is even worse if spilling occurred and we maybe have to page it back in from disk?

I wonder if there's a better way of dealing with this.

Edit: i think this might only be triggered from certain codepaths, and in most cases the metadata is returned as an objectref after execution. This code is so messy...

Doing some experiments in #3557 but it seems like ray.wait(fetch_local=True) doesn't seem to do anything for the metadata in my case (I still hit OOMs). I'm guessing maybe for small objects like the metadata this doesn't actually work, and it only works for things in the object store?

Doing another round of tests now using an explicit ray.get and trying to explicitly set the cached metadata object in the result. If that works, then it could work well for your use-case also.

Doing another round of tests now using an explicit ray.get and trying to explicitly set the cached metadata object in the result. If that works, then it could work well for your use-case also.

Tried this out, essentially always fetch metadata and cache it upon partition task completion, unless it is the final partition task.

Tried this on a few shuffle configurations (1000 x 1000, 2000 x 2000), and it works pretty well, roughly 20s faster.

This could be viable? I believe as long as the task is not final, the partition metadata will need to be retrieved anyway for subsequent tasks, so might as well retrieve it once it is complete. Since it is a blocking call we want to pipeline it with running tasks, which should work if we retrieve it during the task awaiting state.

Made a refactor to have the metadata caching logic on the scheduler instead of on reduce op, ptal!

Made a refactor to have the metadata caching logic on the scheduler instead of on reduce op, ptal! @jaychia

Tested on TPCH SF 1000 on a 8 x i8.4xlarge node cluster.

Not much difference